# -*- coding: utf-8 -*-
from datetime import timedelta
from time import time
from utils.operators.rich_sql_sensor import RichSqlSensor
from spmi.dm.spmi.dm_spmi_composite_cost_weight_sum_dt import spmi_dm__dm_spmi_composite_cost_weight_sum_dt

#doris表名
doris_table = "dm_spmi_composite_cost_weight_sum_dt"
#hive中表名
hive_table = "dm_spmi_composite_cost_weight_sum_dt"
#Task ID
task_id = f"doris_{doris_table}"

label = f"doris_{doris_table}_{int(time())}"
timeout = timedelta(minutes=60).seconds


doris_dm_spmi_composite_cost_weight_sum_dt = RichSqlSensor(
    task_id=f'{task_id}',
    pool='broker_load_pool',
    email=['matthew.xiong@jtexpress.com', 'yl_bigdata@yl-scm.com'],
    conn_id='doris',
    pre_sql=f"""
    TRUNCATE TABLE spmi_dm.{doris_table} PARTITION (p{{{{ execution_date | cst_ds_nodash }}}});
    LOAD LABEL spmi_dm.{label} (
        DATA INFILE(
                    "hdfs://{{{{var.value.hadoop_namespace}}}}/dw/hive/spmi_dm.db/external/{hive_table}/dt={{{{ execution_date | cst_ds }}}}/*"
                    )
        INTO TABLE {doris_table}  
    FORMAT AS 'PARQUET'(
        sign_date
        ,waybill_date
        ,send_network_code
        ,send_network_name
        ,sign_network_code
        ,sign_network_name
        ,receiver_province_id
        ,receiver_province_name
        ,cost_weight1_cnt
        ,cost_weight2_cnt
        ,cost_weight3_cnt
        ,cost_weight4_cnt
        ,cost_weight5_cnt
        ,cost_weight6_cnt
        ,cost_weight7_cnt
        ,cost_weight8_cnt
        ,cost_weight9_cnt
        ,cost_weight10_cnt
        ,cost_weight11_cnt
        ,cost_weight12_cnt
        ,cost_weight13_cnt
        ,cost_weight14_cnt
        ,cost_weight_total_cnt
        ,cost_weight1_sum
        ,cost_weight2_sum
        ,cost_weight3_sum
        ,cost_weight4_sum
        ,cost_weight5_sum
        ,cost_weight6_sum
        ,cost_weight7_sum
        ,cost_weight8_sum
        ,cost_weight9_sum
        ,cost_weight10_sum
        ,cost_weight11_sum
        ,cost_weight12_sum
        ,cost_weight13_sum
        ,cost_weight14_sum
        ,cost_weight_total_sum
        ,cost_weight1_piece_fee_sum
        ,cost_weight2_piece_fee_sum
        ,cost_weight3_piece_fee_sum
        ,cost_weight4_piece_fee_sum
        ,cost_weight5_piece_fee_sum
        ,cost_weight6_piece_fee_sum
        ,cost_weight7_piece_fee_sum
        ,cost_weight8_piece_fee_sum
        ,cost_weight9_piece_fee_sum
        ,cost_weight10_piece_fee_sum
        ,cost_weight11_piece_fee_sum
        ,cost_weight12_piece_fee_sum
        ,cost_weight13_piece_fee_sum
        ,cost_weight14_piece_fee_sum
        ,cost_weight_total_piece_fee_sum
        ,cost_weight1_fee_sum
        ,cost_weight2_fee_sum
        ,cost_weight3_fee_sum
        ,cost_weight4_fee_sum
        ,cost_weight5_fee_sum
        ,cost_weight6_fee_sum
        ,cost_weight7_fee_sum
        ,cost_weight8_fee_sum
        ,cost_weight9_fee_sum
        ,cost_weight10_fee_sum
        ,cost_weight11_fee_sum
        ,cost_weight12_fee_sum
        ,cost_weight13_fee_sum
        ,cost_weight14_fee_sum
        ,cost_weight_fee_sum
   )
   columns from path as (dt)
   set(
         sign_date                          =  sign_date
        ,waybill_date                       =  waybill_date
        ,send_network_code                  =  send_network_code
        ,send_network_name                  =  send_network_name
        ,sign_network_code                  =  sign_network_code
        ,sign_network_name                  =  sign_network_name
        ,receiver_province_id               =  receiver_province_id
        ,receiver_province_name             =  receiver_province_name
        ,cost_weight1_cnt                   =  cost_weight1_cnt
        ,cost_weight2_cnt                   =  cost_weight2_cnt
        ,cost_weight3_cnt                   =  cost_weight3_cnt
        ,cost_weight4_cnt                   =  cost_weight4_cnt
        ,cost_weight5_cnt                   =  cost_weight5_cnt
        ,cost_weight6_cnt                   =  cost_weight6_cnt
        ,cost_weight7_cnt                   =  cost_weight7_cnt
        ,cost_weight8_cnt                   =  cost_weight8_cnt
        ,cost_weight9_cnt                   =  cost_weight9_cnt
        ,cost_weight10_cnt                  =  cost_weight10_cnt
        ,cost_weight11_cnt                  =  cost_weight11_cnt
        ,cost_weight12_cnt                  =  cost_weight12_cnt
        ,cost_weight13_cnt                  =  cost_weight13_cnt
        ,cost_weight14_cnt                  =  cost_weight14_cnt
        ,cost_weight_total_cnt              =  cost_weight_total_cnt
        ,cost_weight1_sum                   =  cost_weight1_sum
        ,cost_weight2_sum                   =  cost_weight2_sum
        ,cost_weight3_sum                   =  cost_weight3_sum
        ,cost_weight4_sum                   =  cost_weight4_sum
        ,cost_weight5_sum                   =  cost_weight5_sum
        ,cost_weight6_sum                   =  cost_weight6_sum
        ,cost_weight7_sum                   =  cost_weight7_sum
        ,cost_weight8_sum                   =  cost_weight8_sum
        ,cost_weight9_sum                   =  cost_weight9_sum
        ,cost_weight10_sum                  =  cost_weight10_sum
        ,cost_weight11_sum                  =  cost_weight11_sum
        ,cost_weight12_sum                  =  cost_weight12_sum
        ,cost_weight13_sum                  =  cost_weight13_sum
        ,cost_weight14_sum                  =  cost_weight14_sum
        ,cost_weight_total_sum              =  cost_weight_total_sum
        ,cost_weight1_piece_fee_sum         =  cost_weight1_piece_fee_sum
        ,cost_weight2_piece_fee_sum         =  cost_weight2_piece_fee_sum
        ,cost_weight3_piece_fee_sum         =  cost_weight3_piece_fee_sum
        ,cost_weight4_piece_fee_sum         =  cost_weight4_piece_fee_sum
        ,cost_weight5_piece_fee_sum         =  cost_weight5_piece_fee_sum
        ,cost_weight6_piece_fee_sum         =  cost_weight6_piece_fee_sum
        ,cost_weight7_piece_fee_sum         =  cost_weight7_piece_fee_sum
        ,cost_weight8_piece_fee_sum         =  cost_weight8_piece_fee_sum
        ,cost_weight9_piece_fee_sum         =  cost_weight9_piece_fee_sum
        ,cost_weight10_piece_fee_sum        =  cost_weight10_piece_fee_sum
        ,cost_weight11_piece_fee_sum        =  cost_weight11_piece_fee_sum
        ,cost_weight12_piece_fee_sum        =  cost_weight12_piece_fee_sum
        ,cost_weight13_piece_fee_sum        =  cost_weight13_piece_fee_sum
        ,cost_weight14_piece_fee_sum        =  cost_weight14_piece_fee_sum
        ,cost_weight_total_piece_fee_sum    =  cost_weight_total_piece_fee_sum
        ,cost_weight1_fee_sum               =  cost_weight1_fee_sum
        ,cost_weight2_fee_sum               =  cost_weight2_fee_sum
        ,cost_weight3_fee_sum               =  cost_weight3_fee_sum
        ,cost_weight4_fee_sum               =  cost_weight4_fee_sum
        ,cost_weight5_fee_sum               =  cost_weight5_fee_sum
        ,cost_weight6_fee_sum               =  cost_weight6_fee_sum
        ,cost_weight7_fee_sum               =  cost_weight7_fee_sum
        ,cost_weight8_fee_sum               =  cost_weight8_fee_sum
        ,cost_weight9_fee_sum               =  cost_weight9_fee_sum
        ,cost_weight10_fee_sum              =  cost_weight10_fee_sum
        ,cost_weight11_fee_sum              =  cost_weight11_fee_sum
        ,cost_weight12_fee_sum              =  cost_weight12_fee_sum
        ,cost_weight13_fee_sum              =  cost_weight13_fee_sum
        ,cost_weight14_fee_sum              =  cost_weight14_fee_sum
        ,cost_weight_fee_sum                =  cost_weight_fee_sum
        ,dt                                 =  dt
        )
    )
    WITH BROKER '{{{{ var.json.doris_brokers | random_choice }}}}'
    PROPERTIES ('timeout'='{timeout}', 'max_filter_ratio'='0.0')""",
    poke_sql=f"SHOW LOAD FROM spmi_dm WHERE label = '{label}' ORDER BY CreateTime DESC LIMIT 1",
    sql_on_kill=f"CANCEL LOAD FROM spmi_dm WHERE LABEL = '{label}'",
    success=lambda r: r[2] == 'FINISHED',
    failure=lambda r: (r[2] is not None and r[2] == 'CANCELLED', str(r[7])),
    poke_interval=60,
    execution_timeout=timedelta(seconds=timeout + 120), )

doris_dm_spmi_composite_cost_weight_sum_dt << spmi_dm__dm_spmi_composite_cost_weight_sum_dt
